package org.apache.ocean;

import java.io.File;
import java.io.Reader;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.logging.Logger;

import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.builder.ReflectionToStringBuilder;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.TokenStream;
import org.apache.lucene.search.DefaultSimilarity;
import org.apache.lucene.search.Similarity;
import org.apache.ocean.analysis.TokenFilterFactory;
import org.apache.ocean.analysis.TokenizerChain;
import org.apache.ocean.analysis.TokenizerFactory;
import org.apache.ocean.name.FileClient;
import org.apache.ocean.server.DateFieldType;
import org.apache.ocean.server.DoubleFieldType;
import org.apache.ocean.server.FieldType;
import org.apache.ocean.server.LongFieldType;
import org.apache.ocean.server.SchemaField;
import org.apache.ocean.server.StringFieldType;
import org.apache.ocean.server.SchemaField.Attribute;
import org.jdom.Element;

/**
 * 
 * @author jasonr
 */
// TODO: verify copy source, and copy to exists
// TODO: test international analyzers
public class IndexSchema {
	public static Logger log = Logger.getLogger(IndexSchema.class.getName());
	private static final String[] packages = { "org.apache.ocean.server", "org.apache.ocean.analysis" };
	public Date lastModifiedDate;
	public Map<String, SchemaField> fieldMap = new HashMap<String, SchemaField>();
	public Map<String, FieldType> fieldTypeMap = new HashMap<String, FieldType>();
	public Map<String, CopyFields> copyFieldMap = new HashMap<String, CopyFields>(); // source
	public Map<String, SchemaField> defaultFieldMap = new HashMap<String, SchemaField>();
	public List<FileClient.FileEventListener> fileEventListeners = new ArrayList<FileClient.FileEventListener>();
	public Analyzer analyzer;
	public Analyzer queryAnalyzer;
	public Similarity similarity = new DefaultSimilarity();
	public String xml;
	public String defaultSearchField;
	public Date lastModified;
	public CLong version;

	public IndexSchema(String xml) throws Exception {
		this.xml = xml;
		Element element = XMLUtil.parseElement(xml);
		init(element);
	}

	public IndexSchema(File schemaFile) throws Exception {
		xml = FileUtils.readFileToString(schemaFile, "UTF-8");
		lastModified = new Date(schemaFile.lastModified());
		Element element = XMLUtil.parseElement(xml);
		init(element);
	}

	public boolean isDefaultField(String name) {
		return defaultFieldMap.containsKey(name);
	}

	private Object newInstance(String name) throws Exception {
		return ClusterUtil.findClass(name, packages).newInstance();
	}

	private void init(Element rootElement) throws Exception {
		version = XMLUtil.getAttributeLong("version", rootElement);
		if (version == null) throw new Exception("version null");
		defaultSearchField = rootElement.getChildText("defaultSearchField");
		Element typesElement = rootElement.getChild("types");
		addFieldType(new DateFieldType());
		addFieldType(new LongFieldType());
		addFieldType(new DoubleFieldType());
		addFieldType(new StringFieldType());
		for (Element typeElement : XMLUtil.getChildren(typesElement)) {
			FieldType fieldType = processFieldType(typeElement);
			addFieldType(fieldType);
		}
		addDefaultFieldTypes();
		Element fieldsElement = rootElement.getChild("fields");
		for (Element fieldElement : XMLUtil.getChildren(fieldsElement)) {
			SchemaField schemaField = processField(fieldElement);
			addSchemaField(schemaField);
		}
		for (Element copyFieldElement : XMLUtil.getChildren("copyField", rootElement)) {
			processCopyField(copyFieldElement);
		}
		analyzer = new OceanIndexAnalyzer();
		queryAnalyzer = new OceanQueryAnalyzer();
	}

	public void addSchemaField(SchemaField schemaField) throws Exception {
		if (fieldMap.containsKey(schemaField.name))
			throw new Exception("schemaField " + schemaField.name + " already defined");
		fieldMap.put(schemaField.name, schemaField);
	}

	private void addDefaultFieldTypes() throws Exception {
		addDefaultFieldType(ClusterConstants.ID_FIELD, "long");
		addDefaultFieldType(ClusterConstants.TRANSACTION_ID_FIELD, "long");
		addDefaultFieldType(ClusterConstants.CELL_TRANSACTION_ID_FIELD, "long");
		addDefaultFieldType(ClusterConstants.SHARD_ID_FIELD, "string");
		addDefaultFieldType(ClusterConstants.VERSION_FIELD, "long");
		addDefaultFieldType(ClusterConstants.ID_VERSION_FIELD, "string");
		addDefaultFieldType(ClusterConstants.GUID_FIELD, "string");
		addDefaultFieldType(ClusterConstants.SCHEMA_VERSION_FIELD, "string");
		addDefaultFieldType(ClusterConstants.TYPE_FIELD, "string");
		addDefaultFieldType(ClusterConstants.CELL_ID_FIELD, "string");
		addDefaultFieldType(ClusterConstants.DATE_CREATED_FIELD, "date");
		addDefaultFieldType(ClusterConstants.DATE_MODIFIED_FIELD, "date");
		addDefaultFieldType(ClusterConstants.LATITUDE_FIELD, "double");
		addDefaultFieldType(ClusterConstants.LONGITUDE_FIELD, "double");
		addDefaultFieldType(ClusterConstants.DISTANCE_FIELD, "double");
	}
  
	public List<String> getUserFieldNames() {
		Set<String> set = new HashSet<String>();
		set.addAll(fieldMap.keySet());
		set.removeAll(defaultFieldMap.keySet());
		set.removeAll(copyFieldMap.keySet());
		return new ArrayList<String>(set);
	}
	
	private void addDefaultFieldType(String name, String type) throws Exception {
		addDefaultFieldType(name, type, true, true);
	}
	
	private void addDefaultFieldType(String name, String type, boolean storeAsPayload, boolean fieldCachePayload) throws Exception {
		FieldType fieldType = fieldTypeMap.get(type);
		SchemaField schemaField = new SchemaField(name, fieldType);
		schemaField.addAttribute(Attribute.INDEXED);
		schemaField.addAttribute(Attribute.STORED);
		if (storeAsPayload) schemaField.addAttribute(Attribute.STOREASPAYLOAD);
		if (fieldCachePayload) schemaField.addAttribute(Attribute.FIELDCACHEPAYLOAD);
		addSchemaField(schemaField);
		defaultFieldMap.put(name, schemaField);
	}

	private void addFieldType(FieldType fieldType) throws Exception {
		if (fieldTypeMap.containsKey(fieldType.getName()))
			throw new Exception(fieldType.getName() + " exists already");
		fieldTypeMap.put(fieldType.getName(), fieldType);
	}

	public Similarity getSimilarity() {
		return similarity;
	}

	public String toXML() {
		return xml;
	}

	public String getDefaultSearchFieldName() {
		return defaultSearchField;
	}

	protected void processCopyField(Element element) throws Exception {
		String source = element.getAttributeValue("source");
		if (!fieldMap.containsKey(source)) {
			throw new Exception("unknown source: "+source);
		}
		CopyFields copyFields = copyFieldMap.get(source);
		if (copyFields == null) {
			copyFields = new CopyFields(source);
			copyFieldMap.put(source, copyFields);
		}
		String destText = element.getAttributeValue("dest");
		for (String dest : ClusterUtil.split(destText)) {
			if (isDefaultField(dest))
				throw new Exception("cannot copy to default field " + dest);
			if (!fieldMap.containsKey(dest)) {
				throw new Exception("unknown dest: "+dest);
			}
			copyFields.addTo(dest);
		}
	}

	protected FieldType processFieldType(Element fieldTypeElement) throws Exception {
		String clazz = fieldTypeElement.getAttributeValue("class");
		String name = fieldTypeElement.getAttributeValue("name");
		FieldType fieldType = createFieldType(clazz);
		fieldType.setTypeName(name);
		for (Element analyzerElement : XMLUtil.getChildren("analyzer", fieldTypeElement)) {
			String type = analyzerElement.getAttributeValue("type");
			Analyzer analyzer = createAnalyzer(analyzerElement);
			if (StringUtils.isBlank(type)) {
				fieldType.setAnalyzer(analyzer);
			} else if (StringUtils.equals("query", type)) {
				fieldType.setQueryAnalyzer(analyzer);
			} else if (StringUtils.equals("index", type)) {
				fieldType.setAnalyzer(analyzer);
			}
		}
		return fieldType;
	}

	protected Analyzer createAnalyzer(Element analyzerElement) throws Exception {
		String analyzerClassName = analyzerElement.getAttributeValue("class");
		if (StringUtils.isNotBlank(analyzerClassName)) {
      return (Analyzer)newInstance(analyzerClassName);
		} else {
			Element tokenizerElement = analyzerElement.getChild("tokenizer");
			TokenizerFactory tokenizerFactory = createTokenizerFactory(tokenizerElement);
			List<TokenFilterFactory> tokenFilterFactories = new ArrayList<TokenFilterFactory>();
			for (Element filterElement : XMLUtil.getChildren("filter", analyzerElement)) {
				TokenFilterFactory tokenFilterFactory = createTokenFilterFactory(filterElement);
				if (tokenFilterFactory instanceof FileClient.FileEventListener) {
					this.fileEventListeners.add((FileClient.FileEventListener)tokenFilterFactory);
				}
				tokenFilterFactories.add(tokenFilterFactory);
			}
			return new TokenizerChain(tokenizerFactory, tokenFilterFactories.toArray(new TokenFilterFactory[tokenFilterFactories.size()]));
		}
	}

	protected TokenizerFactory createTokenizerFactory(Element tokenizerElement) throws Exception {
		String clazz = tokenizerElement.getAttributeValue("class");
		TokenizerFactory tokenizerFactory = (TokenizerFactory) newInstance(clazz);
		Map<String, String> args = ClusterUtil.toMapExcept(tokenizerElement, "class");
		tokenizerFactory.init(args);
		return tokenizerFactory;
	}

	protected TokenFilterFactory createTokenFilterFactory(Element filterElement) throws Exception {
		String clazz = filterElement.getAttributeValue("class");
		TokenFilterFactory tokenFilterFactory = (TokenFilterFactory) newInstance(clazz);
		Map<String, String> args = ClusterUtil.toMapExcept(filterElement, "class");
		tokenFilterFactory.init(args);
		return tokenFilterFactory;
	}

	protected FieldType createFieldType(String className) throws Exception {
		return (FieldType) newInstance(className);
	}

	private void exceptionIfFieldUnknown(String fieldName) {
		if (!fieldMap.containsKey(fieldName)) {
			throw new RuntimeException("unknown field: " + fieldName);
		}
	}

	protected SchemaField processField(Element element) throws Exception {
		String name = element.getAttributeValue("name");
		String type = element.getAttributeValue("type");
		FieldType fieldType = fieldTypeMap.get(type);
		if (fieldType == null)
			throw new Exception("unknown field type: " + type);
		SchemaField schemaField = new SchemaField(name, fieldType);
		schemaField.fieldType = fieldType;
		
		if (XMLUtil.getAttributeBooleanPrimitive("stored", element)) {
			schemaField.addAttribute(Attribute.STORED);
		}
		if (XMLUtil.getAttributeBooleanPrimitive("omitNorms", element)) {
			schemaField.addAttribute(Attribute.OMITNORMS);
		}
		if (XMLUtil.getAttributeBooleanPrimitive("indexed", element)) {
			schemaField.addAttribute(Attribute.INDEXED);
		}
		if (schemaField.name == null)
			throw new Exception("name null " + XMLUtil.printAndDetach(element));
		return schemaField;
	}
  
	public SchemaField getField(String name) {
		return fieldMap.get(name);
	}

	public FieldType getFieldType(String name) {
		return fieldTypeMap.get(name);
	}

	public static class CopyFields {
		public String from;
		public Set<String> tos = new HashSet<String>();

		public CopyFields(String from) {
			this.from = from;
		}

		public void addTo(String to) {
			tos.add(to);
		}

		public String toString() {
			return ReflectionToStringBuilder.toString(this);
		}
	}

	public class OceanIndexAnalyzer extends Analyzer {
		protected final Map<String, Analyzer> analyzerMap = new HashMap<String, Analyzer>();

		public OceanIndexAnalyzer() {
			for (SchemaField schemaField : fieldMap.values()) {
				FieldType fieldType = schemaField.getFieldType();
				if (fieldType == null)
					throw new RuntimeException("fieldType null for " + schemaField);
				Analyzer analyzer = schemaField.getAnalyzer();
				if (analyzer == null) analyzer = fieldType.getAnalyzer();
				analyzerMap.put(schemaField.getName(), analyzer);
			}
		}

		protected Analyzer getAnalyzer(String fieldName) {
			exceptionIfFieldUnknown(fieldName);
			Analyzer analyzer = analyzerMap.get(fieldName);
			if (analyzer == null)
				throw new RuntimeException("no analyzer for field: " + fieldName);
			return analyzer;
		}

		public TokenStream tokenStream(String fieldName, Reader reader) {
			exceptionIfFieldUnknown(fieldName);
			return getAnalyzer(fieldName).tokenStream(fieldName, reader);
		}

		public int getPositionIncrementGap(String fieldName) {
			return getAnalyzer(fieldName).getPositionIncrementGap(fieldName);
		}
	}

	private class OceanQueryAnalyzer extends OceanIndexAnalyzer {
		public OceanQueryAnalyzer() {
			for (SchemaField schemaField : fieldMap.values()) {
				Analyzer analyzer = schemaField.getFieldType().getQueryAnalyzer();
				analyzerMap.put(schemaField.getName(), analyzer);
			}
		}

		protected Analyzer getAnalyzer(String fieldName) {
			exceptionIfFieldUnknown(fieldName);
			Analyzer analyzer = analyzerMap.get(fieldName);
			return analyzer;
		}
	}

	public Analyzer getAnalyzer() {
		return analyzer;
	}

	public Analyzer getQueryAnalyzer() {
		return queryAnalyzer;
	}
}
